package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}

object Code15ReduceByKey extends App {
  val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("Mysql2Text"))

  /**
   * ReduceByKey:
   *      作用在KV形式的RDD上，对Key进行分组操作，要求传入一个 func: (V, V) => V 函数
   *      其中该函数中 参数为相同Key 前后的两个Value值 并且经过计算后，其返回结果类型为Value,该Value作为下一次 函数的第一个参数
   *
   * ReduceByKey可以对数据进行做预聚合：
   *
   *
   */

  // 需求：对班级性别，分组求班级的平均年龄

  sc
    .textFile("scala_code/data/students.txt")

    .map {
      case oneLine => {
        val splitRes: Array[String] = oneLine.split(",")
        //        (splitRes(0),splitRes(1),splitRes(2),splitRes(3),splitRes(4))
        // (id,name,age,gender,clazz)
        ((splitRes(4), splitRes(3)), (1, splitRes(2).toInt))
      }
    }

    //    .groupByKey()
    //    .mapValues {
    //      case iterable => {
    //        var totalAge = 0
    //        var totalNum = 0
    //
    //        iterable
    //          .foreach {
    //            case (id, name, age) => {
    //              totalAge += age.toInt
    //              totalNum += 1
    //            }
    //          }
    //        // 作为结果进行返回出去
    //        totalAge / totalNum.toDouble
    //      }
    //    }
    //    .foreach(println)
    // func: (V, V) => V
    //    .reduceByKey(
    //      (v1: (String, String, Int), v2: (String, String, Int)) => {
    //        (v2._1, v2._2, (v2._3 + v1._3))
    //      }
    //    )
    .reduceByKey(
      (v1: (Int, Int), v2: (Int, Int)) => {
        (v1._1 + v2._1, (v2._2 + v1._2))
      }
    )
    .mapValues {
      case (num, totalAge) => (totalAge / num.toDouble, num)
    }
    .foreach(println)



  // 需求：统计班级各年龄段人数

  sc
    .textFile("scala_code/data/students.txt")

    .map {
      case oneLine => {
        val splitRes: Array[String] = oneLine.split(",")
        //        (splitRes(0),splitRes(1),splitRes(2),splitRes(3),splitRes(4))
        // (id,name,age,gender,clazz)
        ((splitRes(4), splitRes(2)), 1)
      }
    }
//    .foreach(println)
//    .reduceByKey(_ + _)
    .reduceByKey {
      case (value1,value2) => {
        value1 + value2
      }
    }
    .foreach(println)

}
